#include "lp_global_vars.h"
#include "config_parser.h"
double max(double a, double b);
double latency[1000000];
//------------------------------------------------------------------------------
void learner_deliver() {
	int i;
	int ld = lowestDelivered;

	for (i = ld + 1; i < lp_interval_performance->highest_deliverable_inst_id + 1; i++) {
		if (data[(i % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].received) {
			if (!data[(i % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].deliver) {
				if (data[(i % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].packet.pack_id == (i + 1)) {

					if (data[(i % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].packet.p_msgs[0].proposer_id == get_config_info()->self_proposer_id) {

						double l = getMiliSeconds() - data[(i % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].packet.p_msgs[0].mili_seconds;
						lp_interval_performance->interval_latency += l;

						if (lp_interval_performance->total_time > DISCARD && lp_interval_performance->total_time < STOPPING_CONDITION * 1000 - DISCARD) {
							latency[lp_interval_performance->latency_counter] = l;
							lp_interval_performance->latency_counter++;
							lp_interval_performance->latency_sum += l;
							lp_interval_performance->latency_sq_sum += pow(l, 2);
						}

						lp_interval_performance->num_packets_delivered_latency++;
						lp_interval_performance->highest_self_deliverable = data[i % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers)].packet.p_msgs[0].location_index_in_proposer;
					}
					lp_interval_performance->interval_num_packets_delivered++;
					data[(i % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].dec = data[(i % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].packet.pack_id;
					data[(i % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].received = false;
					data[(i % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].deliver = false;

					lowestDelivered = i;
				} else {
					printf("oops! %d %d\n", data[(i % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].packet.pack_id, (i + 1));
					break;
				}
			} else
				exit(-1);
		} else {
			printf("from proposer: %d data is not received!\n", data[(i % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].packet.p_msgs[0].proposer_id);
			break;
		}
	}
}
//------------------------------------------------------------------------------
void recv_new_packet_from_predecessor(int sd, short event, void * arg) {

	UNUSED_ARG(event);
	UNUSED_ARG(arg);

	int status = -1;
	char type;
	status = recv(sd, &type, sizeof(char), MSG_PEEK);
	lp_interval_performance->num_interrupts++;

	if (type == 'i') {
		status = recv(sd, new_data, sizeof(inst), MSG_WAITALL);
		assert(status == sizeof(inst));
		lp_interval_performance->interval_bytes_received += status;
		lp_handel_case_a_inst(new_data);
		lp_interval_performance->num_interrupts_new_msg++;
	} else if (type == 's') {
		status = recv(sd, new_data, sizeof(short_dec_notification), MSG_WAITALL);
		assert(status == sizeof(short_dec_notification));
		lp_interval_performance->interval_bytes_received += status;
		lp_handel_case_b_notification(new_data);
	} else if (type == 'p') {
		status = recv(sd, new_data, sizeof(proposer_msg), MSG_WAITALL);
		assert(status == sizeof(proposer_msg));
		lp_interval_performance->interval_bytes_received += status;
		lp_handel_case_c_pmsg(new_data);
		lp_interval_performance->num_interrupts_new_msg++;
	}
	learner_deliver();

}
//------------------------------------------------------------------------------
void lp_handel_case_a_inst(inst * new_data) {
	// Learner neither has the content nor the decision.(about location)
	// the learner gets both at once.
	int status = -1;

	lp_interval_performance->highest_deliverable_inst_id = new_data->highestDeliverable;
	lp_interval_performance->interval_num_packets_received++;
	lp_interval_performance->total_num_packets_received++;

	memcpy(&data[(new_data->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))], new_data, sizeof(inst));
	assert(data[(new_data->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].instance_id + 1 == data[(new_data->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].packet.pack_id);

	data[(new_data->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].received = true;
	data[(new_data->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].deliver = false;
	data[(new_data->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].highestDeliverable = -1;

	new_data->packet.p_msgs[0].num_nodes_visited++;

	if (new_data->packet.p_msgs[0].num_nodes_visited != get_config_info()->ring_size) {
		status = send(get_config_info()->learner_infos[get_config_info()->self_learner_id].successor_sid, new_data, sizeof(inst), MSG_WAITALL);
		assert(status == sizeof(inst));
		lp_interval_performance->interval_bytes_forwarded += status;
	} else {
		short_dec_notification short_decision;

		short_decision.highestDeliverable = new_data->highestDeliverable;
		short_decision.instance_id = new_data->instance_id;
		short_decision.pack_id = new_data->packet.pack_id;
		assert(short_decision.instance_id + 1 == short_decision.pack_id);
		short_decision.proposer_id = new_data->packet.p_msgs[0].proposer_id;
		short_decision.location_index_in_proposer = new_data->packet.p_msgs[0].location_index_in_proposer;
		short_decision.highestDeliverable = lp_interval_performance->highest_deliverable_inst_id;
		short_decision.proposer_msg_id = new_data->packet.p_msgs[0].msg_id;
		short_decision.type = 's';

		status = send(get_config_info()->learner_infos[get_config_info()->self_learner_id].successor_sid, &short_decision, sizeof(short_dec_notification), MSG_WAITALL);
		assert(status == sizeof(short_dec_notification));
		lp_interval_performance->interval_bytes_forwarded += status;
	}
}
//------------------------------------------------------------------------------
void lp_handel_case_b_notification(inst * new_data) {
	// Learner has the message somewhere. Gets the final location.
	int status = -1;
	short_dec_notification *short_decision = (short_dec_notification*) new_data;
	assert(lp_interval_performance->highest_deliverable_inst_id <= short_decision->highestDeliverable);
	lp_interval_performance->highest_deliverable_inst_id = short_decision->highestDeliverable;
	assert(p_msgs[short_decision->proposer_id][(short_decision->location_index_in_proposer) % INTERNAL_BUFFER_SIZE].msg_id == short_decision->proposer_msg_id);
	if (short_decision->instance_id != -1) {

		p_msgs[short_decision->proposer_id][(short_decision->location_index_in_proposer) % INTERNAL_BUFFER_SIZE].final_decision = short_decision->instance_id;
		data[(short_decision->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].received = true;
		data[(short_decision->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].instance_id = short_decision->instance_id;
		data[(short_decision->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].packet.pack_id = short_decision->pack_id;
		assert(data[(short_decision->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].packet.pack_id == data[(short_decision->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].instance_id + 1);

		data[(short_decision->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].deliver = false;
		data[(short_decision->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].packet.p_msgs[0].mili_seconds = p_msgs[short_decision->proposer_id][(short_decision->location_index_in_proposer) % INTERNAL_BUFFER_SIZE].mili_seconds;

		assert(short_decision->location_index_in_proposer == p_msgs[short_decision->proposer_id][(short_decision->location_index_in_proposer) % INTERNAL_BUFFER_SIZE].location_index_in_proposer);

		data[(short_decision->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].packet.p_msgs[0].location_index_in_proposer = short_decision->location_index_in_proposer;
		data[(short_decision->instance_id % (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers))].packet.p_msgs[0].proposer_id = short_decision->proposer_id;
	} else {
		printf("this is the case : %d	index_prop: %d 	p_id: %d	\n", short_decision->highestDeliverable, short_decision->location_index_in_proposer, short_decision->proposer_id);
		exit(-1);
	}
	if (get_config_info()->self_learner_id != get_config_info()->total_num_learners - 1) {
		status = send(get_config_info()->learner_infos[get_config_info()->self_learner_id].successor_sid, short_decision, sizeof(short_dec_notification), MSG_WAITALL);
		assert(status == sizeof(short_dec_notification));
		lp_interval_performance->interval_bytes_forwarded += status;
	}
}
//------------------------------------------------------------------------------
void lp_handel_case_c_pmsg(inst * new_data) {
	//learner gets the message content and inserts it in a buffer.
	int status = -1;
	proposer_msg *p_msg = (proposer_msg*) new_data;

	memcpy(&p_msgs[p_msg->proposer_id][(p_msg->location_index_in_proposer) % INTERNAL_BUFFER_SIZE], p_msg, sizeof(proposer_msg));
	p_msg->num_nodes_visited++;

	double max_hd = max(p_msg->highest_deliverable_inst_id, lp_interval_performance->highest_deliverable_inst_id);
	lp_interval_performance->highest_deliverable_inst_id = max_hd;
	p_msg->highest_deliverable_inst_id = max_hd;

	status = send(get_config_info()->learner_infos[get_config_info()->self_learner_id].successor_sid, p_msg, sizeof(proposer_msg), MSG_WAITALL);
	//	printf("status: %d\n",status);
	assert(status == sizeof(proposer_msg));
	lp_interval_performance->interval_bytes_forwarded += status;

}
//------------------------------------------------------------------------------
double max(double a, double b) {
	if (a >= b)
		return a;
	else
		return b;
}
